# RocketMQ 功能特性

## 1. 概述

本文介绍了 RocketMQ 各类功能特性和其常见的应用场景，以及 Java 的代码实现

## 2. 普通消息

普通消息为 Apache RocketMQ 中最基础的消息，区别于有特性的顺序消息、定时/延时消息和事务消息。本文为您介绍普通消息的应用场景、功能原理、使用方法和使用建议。

### 应用场景

- 微服务异步解耦 - 事件消息封装&消息订阅 Task 处理
- 数据集成传输 - 数据转发 & （分类存储 | 分类分析）

### 功能原理

**什么是普通消息**

定义：普通消息是 Apache RocketMQ 基本消息功能，支持生产者和消费者的异步解耦通信。

**普通消息生命周期**

- 初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。
- 待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。
- 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ 会对消息进行重试处理。具体信息，请参见[消费重试](https://rocketmq.apache.org/zh/docs/featureBehavior/10consumerretrypolicy)。
- 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。 Apache RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
- 消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。更多信息，请参见[消息存储和清理机制](https://rocketmq.apache.org/zh/docs/featureBehavior/11messagestorepolicy)。

### 使用限制

普通消息仅支持使用 MessageType 为 Normal 主题，即普通消息只能发送至类型为普通消息的主题中，发送的消息的类型必须和主题的类型一致。

### 使用示例

**创建主题**

Apache RocketMQ 5.0 版本下创建主题操作，推荐使用 mqadmin 工具，需要注意的是，对于消息类型需要通过属性参数添加。示例如下：

```bash
sh mqadmin updatetopic -t topic -c DefaultCluster -a +message.type=NORMAL
```

**常用命令**

```bash
# 创建主题
sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=[NORMAL|ORDERLY|TRANSACTION|DELAY|FIFO]
# 查看所有topic
sh mqadmin topicList
# 通过messageKey查看topic下的消息
sh mqadmin queryMsgByKey  -t <topic_name> -k <key_name>
# 查询主题下的消息队列
sh mqadmin topicStatus -t <topic_name>
# 删除topic
sh mqadmin deleteTopic -c <cluster-name> -t <topic_name>
```

**发送消息** 普通消息支持设置消息索引键、消息过滤标签等信息，用于消息过滤和搜索查找。以 Java 语言为例，收发普通消息的示例代码如下：

```java
# https://gitee.com/leo-blaze/examples/tree/master/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/NormalMessage.java
public static void main(String[] args) throws ClientException, IOException, InterruptedException {
  // sendNormalMessage(); // 发送消息到topic
  // pushConsumer(); // 接收消息从topic
  // simpleConsumer(); // 接收消息从topic
}
```

## 3. 定时/延时消息

定时/延时消息为 Apache RocketMQ 中的高级特性消息，本文为您介绍定时/延时消息的应用场景、功能原理、使用限制、使用方法和使用建议。

<Alert
  message="备注"
  description="定时消息和延时消息本质相同，都是服务端根据消息设置的定时时间在某一固定时刻将消息投递给消费者消费。因此，下文统一用定时消息描述。"
  type="info"
  showIcon
/>

### 应用场景

- 分布式定时调度
- 任务超时处理

### 功能原理

**什么是定时消息**

定时消息是 Apache RocketMQ 提供的一种高级消息类型，消息被发送至服务端后，在指定时间后才能被消费者消费。通过设置一定的定时时间可以实现分布式场景的延时调度触发效果。

**定时时间设置原则**

- Apache RocketMQ 定时消息设置的定时时间是一个预期触发的系统时间戳，延时时间也需要转换成当前系统时间后的某一个时间戳，而不是一段延时时长。
- 定时时间的格式为毫秒级的 Unix 时间戳，您需要将要设置的时刻转换成时间戳形式。具体方式，请参见 [Unix 时间戳转换工具](https://www.unixtimestamp.com/)
- 定时时间必须设置在定时时长范围内，超过范围则定时不生效，服务端会立即投递消息。
- 定时时长最大值默认为 24 小时，不支持自定义修改，更多信息，请参见[参数限制](https://rocketmq.apache.org/zh/docs/introduction/03limits)。
- 定时时间必须设置为当前时间之后，若设置到当前时间之前，则定时不生效，服务端会立即投递消息。

**定时消息生命周期**

- 初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。
- 定时中：消息被发送到服务端，和普通消息不同的是，服务端不会直接构建消息索引，而是会将定时消息单独存储在定时存储系统中，等待定时时刻到达。
- 待消费：定时时刻到达后，服务端将消息重新写入普通存储引擎，对下游消费者可见，等待消费者消费的状态。
- 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ 会对消息进行重试处理。具体信息，请参见[消费重试](https://rocketmq.apache.org/zh/docs/featureBehavior/10consumerretrypolicy)。
- 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。 Apache RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
- 消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。更多信息，请参见[消息存储和清理机制](https://rocketmq.apache.org/zh/docs/featureBehavior/11messagestorepolicy)。

### 使用限制

**消息类型一致性**

定时消息仅支持在 MessageType 为 Delay 的主题内使用，即定时消息只能发送至类型为定时消息的主题中，发送的消息的类型必须和主题的类型一致。

**定时精度约束**

Apache RocketMQ 定时消息的定时时长参数精确到毫秒级，但是默认精度为 1000ms，即定时消息为秒级精度。

Apache RocketMQ 定时消息的状态支持持久化存储，系统由于故障重启后，仍支持按照原来设置的定时时间触发消息投递。若存储系统异常重启，可能会导致定时消息投递出现一定延迟。

### 使用示例

**创建主题**

```bash
sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=DELAY
```

**发送消息**

和普通消息相比，定时消费发送时，必须设置定时触发的目标时间戳。

```java
#  sh mqadmin updatetopic -t DelayTopic -c DefaultCluster -a +message.type=DELAY
# https://gitee.com/leo-blaze/examples/tree/master/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/DelayMessage.java
  public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    // sendDelayMessage(); // 发送消息到topic
    // pushConsumer(); // 接收消息从topic
    // simpleConsumer(); // 接收消息从topic
  }
```

### 使用建议

**避免大量相同定时时刻的消息**

定时消息的实现逻辑需要先经过定时存储等待触发，定时时间到达后才会被投递给消费者。因此，如果将大量定时消息的定时时间设置为同一时刻，则到达该时刻后会有大量消息同时需要被处理，会造成系统压力过大，导致消息分发延迟，影响定时精度。

## 4. 顺序消息

顺序消息为 Apache RocketMQ 中的高级特性消息，本文为您介绍顺序消息的应用场景、功能原理、使用限制、使用方法和使用建议。

### 应用场景

- 撮合交易
- 数据实时增量同步

### 功能原理

**什么是顺序消息**

顺序消息是 Apache RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理。 相比其他类型消息，顺序消息在发送、存储和投递的处理过程中，更多强调多条消息间的先后顺序关系。

Apache RocketMQ 顺序消息的顺序关系通过消息组（MessageGroup）判定和识别，发送顺序消息时需要为每条消息设置归属的消息组，相同消息组的多条消息之间遵循先进先出的顺序关系，不同消息组、无消息组的消息之间不涉及顺序性。

基于消息组的顺序判定逻辑，支持按照业务逻辑做细粒度拆分，可以在满足业务局部顺序的前提下提高系统的并行度和吞吐能力。

**如何保证消息的顺序性**

Apache RocketMQ 的消息的顺序性分为两部分，生产顺序性和消费顺序性。

- **生产顺序性 ：**

  Apache RocketMQ 通过生产者和服务端的协议保障单个生产者串行地发送消息，并按序存储和持久化。

  如需保证消息生产的顺序性，则必须满足以下条件：

  - 投递顺序

        Apache RocketMQ 通过客户端 SDK 和服务端通信协议保障消息按照服务端存储顺序投递，但业务方消费消息时需要严格按照接收---处理---应答的语义处理消息，避免因异步处理导致消息乱序。

        <Alert

    message="备注"
    description="消费者类型为 PushConsumer 时， Apache RocketMQ 保证消息按照存储顺序一条一条投递给消费者，若消费者类型为 SimpleConsumer，则消费者有可能一次拉取多条消息。此时，消息消费的顺序性需要由业务方自行保证。消费者类型的具体信息，请参见消费者分类。"
    type="info"
    showIcon
    />

  - 有限重试

    Apache RocketMQ 顺序消息投递仅在重试次数限定范围内，即一条消息如果一直重试失败，超过最大重试次数后将不再重试，跳过这条消息消费，不会一直阻塞后续消息处理。

    对于需要严格保证消费顺序的场景，请务设置合理的重试次数，避免参数不合理导致消息乱序。

**生产顺序性和消费顺序性组合**

如果消息需要严格按照先进先出（FIFO）的原则处理，即先发送的先消费、后发送的后消费，则必须要同时满足生产顺序性和消费顺序性。

一般业务场景下，同一个生产者可能对接多个下游消费者，不一定所有的消费者业务都需要顺序消费，您可以将生产顺序性和消费顺序性进行差异化组合，应用于不同的业务场景。例如发送顺序消息，但使用非顺序的并发消费方式来提高吞吐能力。更多组合方式如下表所示：

| 生产顺序                    | 消费顺序 | 顺序性效果                                                                            |
| --------------------------- | -------- | ------------------------------------------------------------------------------------- |
| ✅ 设置消息组，保证顺序发送 | 顺序消费 | 🔒 **严格顺序保证**：按照消息组粒度，同一消息组内的消息消费顺序和发送顺序完全一致     |
| ✅ 设置消息组，保证顺序发送 | 并发消费 | ⚡ **尽力有序**：并发消费，尽可能按时间顺序处理                                       |
| ❌ 未设置消息组，乱序发送   | 顺序消费 | 📊 **队列存储顺序**：基于队列属性，消费顺序和队列存储顺序一致，但不保证和发送顺序一致 |
| ❌ 未设置消息组，乱序发送   | 并发消费 | 🔄 **完全并发**：并发消费，尽可能按照时间顺序处理                                     |

**顺序消息生命周期**

- 初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。
- 待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。
- 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ 会对消息进行重试处理。具体信息，请参见[消费重试](https://rocketmq.apache.org/zh/docs/featureBehavior/10consumerretrypolicy)。
- 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。 Apache RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
- 消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。更多信息，请参见消息[存储和清理机制](https://rocketmq.apache.org/zh/docs/featureBehavior/11messagestorepolicy)。

<Alert
  description="消息消费失败或消费超时，会触发服务端重试逻辑，重试消息属于新的消息，原消息的生命周期已结束。"
  type="info"
  showIcon
/>
<Alert
  description="顺序消息消费失败进行消费重试时，为保障消息的顺序性，后续消息不可被消费，必须等待前面的消息消费完成后才能被处理。"
  type="info"
  showIcon
/>

### 使用限制

顺序消息仅支持使用 MessageType 为 FIFO 的主题，即顺序消息只能发送至类型为顺序消息的主题中，发送的消息的类型必须和主题的类型一致。

### 使用示例

**创建主题**

Apache RocketMQ 5.0 版本下创建主题操作，推荐使用 mqadmin 工具，需要注意的是，对于消息类型需要通过属性参数添加。示例如下：

```bash
sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=FIFO
```

**创建订阅消费组**

Apache RocketMQ 5.0 版本下创建订阅消费组操作，推荐使用 mqadmin 工具，需要注意的是，对于订阅消费组顺序类型需要通过 -o 选项设置。示例如下：

```bash
sh mqadmin updateSubGroup -c <cluster_name> -g <consumer_group_name> -n <nameserver_address> -o true
```

**发送消息**

和普通消息发送相比，顺序消息发送必须要设置消息组。消息组的粒度建议按照业务场景，尽可能细粒度设计，以便实现业务拆分和并发扩展。

```java
# 创建主题
sh mqadmin updatetopic -t FIFOTopic -c DefaultCluster -a +message.type=FIFO
# 创建订阅消费组
sh mqadmin updateSubGroup -c DefaultCluster -g FIFOGroup -o true

# https://gitee.com/leo-blaze/examples/tree/master/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/FIFOMessage.java
  public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    // sendFIFOMessage(); // 发送消息到topic
    // pushConsumer(); // 接收消息从topic
    // simpleConsumer(); // 接收消息从topic
  }
```

### 使用建议

**串行消费，避免批量消费导致乱序**

消息消费建议串行处理，避免一次消费多条消息，否则可能出现乱序情况。

例如：发送顺序为 1->2->3->4，消费时批量消费，消费顺序为 1->23（批量处理，失败）->23（重试处理）->4，此时可能由于消息 3 的失败导致消息 2 被重复处理，最后导致消息消费乱序。

**消息组尽可能打散，避免集中导致热点**

Apache RocketMQ 保证相同消息组的消息存储在同一个队列中，如果不同业务场景的消息都集中在少量或一个消息组中，则这些消息存储压力都会集中到服务端的少量队列或一个队列中。容易导致性能热点，且不利于扩展。一般建议的消息组设计会采用订单 ID、用户 ID 作为顺序参考，即同一个终端用户的消息保证顺序，不同用户的消息无需保证顺序。

因此建议将业务以消息组粒度进行拆分，例如，将订单 ID、用户 ID 作为消息组关键字，可实现同一终端用户的消息按照顺序处理，不同用户的消息无需保证顺序。

## 5. 事务消息

事务消息为 Apache RocketMQ 中的高级特性消息，本文为您介绍事务消息的应用场景、功能原理、使用限制、使用方法和使用建议。

### 应用场景

- 分布式事务的诉求

  分布式系统调用的特点为一个核心业务逻辑的执行，同时需要调用多个下游业务进行处理。因此，如何保证核心业务和多个下游业务的执行结果完全一致，是分布式事务需要解决的主要问题。

- 传统 XA 事务方案：性能不足

  为了保证上述四个分支的执行结果一致性，典型方案是基于 XA 协议的分布式事务系统来实现。将四个调用分支封装成包含四个独立事务分支的大事务。基于 XA 分布式事务的方案可以满足业务处理结果的正确性，但最大的缺点是多分支环境下资源锁定范围大，并发度低，随着下游分支的增加，系统性能会越来越差。

- 基于普通消息方案：一致性保障困难

  将上述基于 XA 事务的方案进行简化，将订单系统变更作为本地事务，剩下的系统变更作为普通消息的下游来执行，事务分支简化成普通消息+订单表事务，充分利用消息异步化的能力缩短链路，提高并发度。

- 基于 Apache RocketMQ 分布式事务消息：支持最终一致性

  上述普通消息方案中，普通消息和订单事务无法保证一致的原因，本质上是由于普通消息无法像单机数据库事务一样，具备提交、回滚和统一协调的能力。

  而基于 Apache RocketMQ 实现的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。

Apache RocketMQ 事务消息的方案，具备高性能、可扩展、业务开发简单的优势。具体事务消息的原理和流程，请参见下文的功能原理。

### 功能原理

**什么是事务消息**

事务消息是 Apache RocketMQ 提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。

**事务消息处理流程**

两阶段提交，详情请查看[官方文档](https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage)

**事务消息生命周期**

- 初始化：半事务消息被生产者构建并完成初始化，待发送到服务端的状态。
- 事务待提交：半事务消息被发送到服务端，和普通消息不同，并不会直接被服务端持久化，而是会被单独存储到事务存储系统中，等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。
- 消息回滚：第二阶段如果事务执行结果明确为回滚，服务端会将半事务消息回滚，该事务消息流程终止。
- 提交待消费：第二阶段如果事务执行结果明确为提交，服务端会将半事务消息重新存储到普通存储系统中，此时消息对下游消费者可见，等待被消费者获取并消费。
- 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ 会对消息进行重试处理。具体信息，请参见[消费重试](https://rocketmq.apache.org/zh/docs/featureBehavior/10consumerretrypolicy)。
- 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。 Apache RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
- 消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。更多信息，请参见[消息存储和清理机制](https://rocketmq.apache.org/zh/docs/featureBehavior/11messagestorepolicy)。

### 使用限制

**消息类型一致性**

事务消息仅支持在 MessageType 为 Transaction 的主题内使用，即事务消息只能发送至类型为事务消息的主题中，发送的消息的类型必须和主题的类型一致。

**消费事务性**

Apache RocketMQ 事务消息保证本地主分支事务和下游消息发送事务的一致性，但不保证消息消费结果和上游事务的一致性。因此需要下游业务分支自行保证消息正确处理，建议消费端做好消费重试，如果有短暂失败可以利用重试机制保证最终处理成功。

**中间状态可见性**

Apache RocketMQ 事务消息为最终一致性，即在消息提交到下游消费端处理完成之前，下游分支和上游事务之间的状态会不一致。因此，事务消息仅适合接受异步执行的事务场景。

**事务超时机制**

Apache RocketMQ 事务消息的生命周期存在超时机制，即半事务消息被生产者发送服务端后，如果在指定时间内服务端无法确认提交或者回滚状态，则消息默认会被回滚。事务超时时间，请参见[参数限制](https://rocketmq.apache.org/zh/docs/introduction/03limits)。

### 使用示例

```java
# 创建主题
sh mqadmin updateTopic -t TRANSACTIONTopic -c DefaultCluster -a +message.type=TRANSACTION

# https://gitee.com/leo-blaze/examples/tree/master/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/TranactionMessage.java
  public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    sendTranactionMessage(); // 执行事务消息
  }
```

### 使用建议

**避免大量未决事务导致超时**

Apache RocketMQ 支持在事务提交阶段异常的情况下发起事务回查，保证事务一致性。但生产者应该尽量避免本地事务返回未知结果。大量的事务检查会导致系统性能受损，容易导致事务处理延迟。

**正确处理"进行中"的事务**

消息回查时，对于正在进行中的事务不要返回 Rollback 或 Commit 结果，应继续保持 Unknown 的状态。 一般出现消息回查时事务正在处理的原因为：事务执行较慢，消息回查太快。解决方案如下：

- 将第一次事务回查时间设置较大一些，但可能导致依赖回查的事务提交延迟较大。
- 程序能正确识别正在进行中的事务。

## 6. 消息发送重试和流控机制

本文为您介绍 Apache RocketMQ 的消息发送重试机制和消息流控机制。

### 重试流程

生产者在初始化时设置消息发送最大重试次数，生产者客户端会按照设置的重试次数一直重试发送消息，直到消息发送成功或达到最大重试次数重试结束，并在最后一次重试失败后返回调用错误响应。

- 同步发送：调用线程会一直阻塞，直到某次重试成功或最终重试失败，抛出错误码和异常。
- 异步发送：调用线程不会阻塞，但调用结果会通过异常事件或者成功事件返回。

### 重试间隔

- 除服务端返回系统流控错误场景，其他触发条件触发重试后，均会立即进行重试，无等待间隔。
- 若由于服务端返回流控错误触发重试，系统会按照指数退避策略进行延迟重试。指数退避算法通过以下参数控制重试行为：
  - INITIAL_BACKOFF： 第一次失败重试前后需等待多久，默认值：1 秒。
  - MULTIPLIER ：指数退避因子，即退避倍率，默认值：1.6。
  - JITTER ：随机抖动因子，默认值：0.2。
  - MAX_BACKOFF ：等待间隔时间上限，默认值：120 秒
  - MIN_CONNECT_TIMEOUT ：最短重试间隔，默认值：20 秒。

**建议算法如下：**

```java
ConnectWithBackoff()
  current_backoff = INITIAL_BACKOFF
  current_deadline = now() + INITIAL_BACKOFF
  while (TryConnect(Max(current_deadline, now() + MIN_CONNECT_TIMEOUT))!= SUCCESS)
    SleepUntil(current_deadline)
    current_backoff = Min(current_backoff * MULTIPLIER, MAX_BACKOFF)
    current_deadline = now() + current_backoff + UniformRandom(-JITTER * current_backoff, JITTER * current_backoff)
```

更多信息，请参见[connection-backoff](https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md) 策略。

### 功能约束

- 链路耗时阻塞评估：从上述重试机制可以看出，在重试流程中生产者仅能控制最大重试次数。若由于系统异常触发了 SDK 内置的重试逻辑，则服务端需要等待最终重试结果，可能会导致消息发送请求链路被阻塞。对于某些实时调用类场景，您需要合理评估每次调用请求的超时时间以及最大重试次数，避免影响全链路的耗时。

- 最终异常兜底： Apache RocketMQ 客户端内置的发送请求重试机制并不能保证消息发送一定成功。当最终重试仍然失败时，业务方调用需要捕获异常，并做好冗余保护处理，避免消息发送结果不一致。

- 消息重复问题：因远程调用的不确定性，当 Apache RocketMQ 客户端因请求超时触发消息发送重试流程，此时客户端无法感知服务端的处理结果，客户端进行的消息发送重试可能会产生消息重复问题，业务逻辑需要自行处理消息重复问题。

### 消息流控机制

#### 消息流控基本概念

消息流控指的是系统容量或流量过高， Apache RocketMQ 服务端会通过快速失败返回流控错误来避免底层资源承受过高压力。

#### 触发条件

Apache RocketMQ 的消息流控触发条件如下：

- 存储压力大：参考消费进度管理的原理机制，消费者分组的初始消费位点为当前队列的最大消费位点。若某些场景例如业务上新等需要回溯到指定时刻前开始消费，此时队列的存储压力会瞬间飙升，触发消息流控
- 服务端请求任务排队溢出：若消费者消费能力不足，导致队列中有大量堆积消息，当堆积消息超过一定数量后会触发消息流控，减少下游消费系统压力。

#### 流控行为

当系统触发消息发送流控时，客户端会收到系统限流错误和异常，错误码信息如下：

- reply-code：530
- reply-text：TOO_MANY_REQUESTS

客户端收到系统流控错误码后，会根据指数退避策略进行消息发送重试。

### 处理建议

- 如何避免触发消息流控：触发限流的根本原因是系统容量或水位过高，您可以利用可观测性功能监控系统水位容量等，保证底层资源充足，避免触发流控机制。
- 突发消息流控处理：如果因为突发原因触发消息流控，且客户端内置的重试流程执行失败，则建议业务方将请求调用临时替换到其他系统进行应急处理。

## 7. 消费者分类

Apache RocketMQ 支持 PushConsumer 、 SimpleConsumer 以及 PullConsumer 这三种类型的消费者，本文分别从使用方式、实现原理、可靠性重试和适用场景等方面为您介绍这三种类型的消费者。

### 应用场景

- 实现并发消费
- 实现同步、异步消息处理
- 实现消息可靠处理

Apache RocketMQ 的消费者处理消息时主要经过以下阶段：消息获取--->消息处理--->消费状态提交。

针对以上几个阶段，Apache RocketMQ 提供了不同的消费者类型： PushConsumer 、SimpleConsumer 和 PullConsumer。这几种类型的消费者通过不同的实现方式和接口可满足您在不同业务场景下的消费需求。具体差异如下：

{" "}

<Alert
  description="在实际使用场景中，PullConsumer 仅推荐在流处理框架中集成使用，大多数消息收发场景使用 PushConsumer 和 SimpleConsumer 就可以满足需求。"
  type="info"
  showIcon
/>
<Alert
  description="若您的业务场景发生变更，或您当前使用的消费者类型不适合当前业务，您可以选择在 PushConsumer 和SimpleConsumer 之间变更消费者类型。变更消费者类型不影响当前Apache RocketMQ 资源的使用和业务处理。"
  type="info"
  showIcon
/>
<Alert
  description="生产环境中相同的 ConsumerGroup 下严禁混用 PullConsumer 和其他两种消费者，否则会导致消息消费异常。"
  type="error"
  showIcon
/>

| 对比项       | PushConsumer    | SimpleConsumer   | PullConsumer    |
| ------------ | --------------- | ---------------- | --------------- |
| **接口方式** | 🔄 回调监听器   | ⚡ 主动 API 调用 | 🎯 队列拉取     |
| **并发管理** | 🤖 SDK 自动管理 | 👨‍💻 业务方管理    | 👨‍💻 业务方管理   |
| **负载均衡** | 📊 消息粒度均衡 | 📊 消息粒度均衡  | 📦 队列粒度批量 |
| **灵活度**   | 🎁 封装度高     | 🔧 高度灵活      | 🔧 高度灵活     |
| **使用难度** | 🟢 简单         | 🟡 中等          | 🔴 复杂         |
| **推荐指数** | ⭐⭐⭐⭐⭐      | ⭐⭐⭐⭐         | ⭐⭐            |
| **适用场景** | 常规业务开发    | 定制化业务       | 流处理框架      |

### PushConsumer

PushConsumers 是一种高度封装的消费者类型，消费消息仅通过消费监听器处理业务并返回消费结果。消息的获取、消费状态提交以及消费重试都通过 Apache RocketMQ 的客户端 SDK 完成。

使用方式

PushConsumer 的使用方式比较固定，在消费者初始化时注册一个消费监听器，并在消费监听器内部实现消息处理逻辑。由 Apache RocketMQ 的 SDK 在后台完成消息获取、触发监听器调用以及进行消息重试处理。

示例代码：[NormalMessage.java](https://gitee.com/leo-blaze/examples/tree/master/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/NormalMessage.java)

PushConsumer 的消费监听器执行结果分为以下三种情况：

- 返回消费成功：以 Java SDK 为例，返回 ConsumeResult.SUCCESS，表示该消息处理成功，服务端按照消费结果更新消费进度。
- 返回消费失败：以 Java SDK 为例，返回 ConsumeResult.FAILURE，表示该消息处理失败，需要根据消费重试逻辑判断是否进行重试消费。
- 出现非预期失败：例如抛异常等行为，该结果按照消费失败处理，需要根据消费重试逻辑判断是否进行重试消费。

PushConsumer 消费消息时，若消息处理逻辑出现预期之外的阻塞导致消息处理一直无法执行成功，SDK 会按照消费超时处理强制提交消费失败结果，并按照消费重试逻辑进行处理。消息超时，请参见 PushConsumer 消费重试策略。

<Alert
  description="出现消费超时情况时，SDK虽然提交消费失败结果，但是当前消费线程可能仍然无法响应中断，还会继续处理消息。"
  type="info"
  showIcon
/>

**内部原理**

在 PushConsumer 类型中，消息的实时处理能力是基于 SDK 内部的典型 Reactor 线程模型实现的。如下图所示，SDK 内置了一个长轮询线程，先将消息异步拉取到 SDK 内置的缓存队列中，再分别提交到消费线程中，触发监听器执行本地消费逻辑。

**可靠性重试**

PushConsumer 消费者类型中，客户端 SDK 和消费逻辑的唯一边界是消费监听器接口。客户端 SDK 严格按照监听器的返回结果判断消息是否消费成功，并做可靠性重试。所有消息必须以同步方式进行消费处理，并在监听器接口结束时返回调用结果，不允许再做异步化分发。消息重试具体信息，请参见 PushConsumer 消费重试策略。

使用 PushConsumer 消费者消费时，不允许使用以下方式处理消息，否则 Apache RocketMQ 无法保证消息的可靠性。

<Alert
  description="错误方式一：消息还未处理完成，就提前返回消费成功结果。此时如果消息消费失败，Apache RocketMQ 服务端是无法感知的，因此不会进行消费重试。"
  type="error"
  showIcon
/>

<Alert
  description="错误方式二：在消费监听器内将消息再次分发到自定义的其他线程，消费监听器提前返回消费结果。此时如果消息消费失败，Apache RocketMQ 服务端同样无法感知，因此也不会进行消费重试。"
  type="error"
  showIcon
/>

**顺序性保障**

基于 Apache RocketMQ 顺序消息的定义，如果消费者分组设置了顺序消费模式，则 PushConsumer 在触发消费监听器时，严格遵循消息的先后顺序。业务处理逻辑无感知即可保证消息的消费顺序。

<Alert
  description="消息消费按照顺序处理的前提是遵循同步提交原则，如果业务逻辑自定义实现了异步分发，则Apache RocketMQ 无法保证消息的顺序性。"
  type="info"
  showIcon
/>

**适用场景**

PushConsumer 严格限制了消息同步处理及每条消息的处理超时时间，适用于以下场景：

- 消息处理时间可预估：如果不确定消息处理耗时，经常有预期之外的长时间耗时的消息，PushConsumer 的可靠性保证会频繁触发消息重试机制造成大量重复消息。
- 无异步化、高级定制场景：PushConsumer 限制了消费逻辑的线程模型，由客户端 SDK 内部按最大吞吐量触发消息处理。该模型开发逻辑简单，但是不允许使用异步化和自定义处理流程。

### SimpleConsumer

SimpleConsumer 是一种接口原子型的消费者类型，消息的获取、消费状态提交以及消费重试都是通过消费者业务逻辑主动发起调用完成。

**使用方式**

SimpleConsumer 的使用涉及多个接口调用，由业务逻辑按需调用接口获取消息，然后分发给业务线程处理消息，最后按照处理的结果调用提交接口，返回服务端当前消息的处理结果。示例如下：

```java
// 消费示例：使用 SimpleConsumer 消费普通消息，主动获取消息处理并提交。
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "YourTopic";
FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG);
SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // 设置消费者分组。
        .setConsumerGroup("YourConsumerGroup")
        // 设置接入点。
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())
        // 设置预绑定的订阅关系。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // 设置从服务端接受消息的最大等待时间
        .setAwaitDuration(Duration.ofSeconds(1))
        .build();
try {
    // SimpleConsumer 需要主动获取消息，并处理。
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // 消费处理完成后，需要主动调用 ACK 提交消费结果。
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            logger.error("Failed to ack message, messageId={}", messageView.getMessageId(), e);
        }
    });
} catch (ClientException e) {
    // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
    logger.error("Failed to receive message", e);
}
```

SimpleConsumer 主要涉及以下几个接口行为：

| 接口名称                    | 主要作用                                                                                                                                                                                                        | 可修改参数                                                                                                                                           |
| --------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- |
| **ReceiveMessage**          | 📥 **主动拉取消息**<br />消费者主动调用该接口从服务端获取消息。<br /><br />⚠️ **注意**：由于服务端存储为分布式，可能会出现服务端实际有消息，但是返回为空的现象。<br />🛠️ **解决方案**：重新发起调用或提高并发度 | 🔢 **批量拉取消息数**<br />一次性批量获取多条消息实现批量消费<br /><br />⏱️ **消费不可见时间**<br />消息的最长处理耗时，控制消费失败时的消息重试间隔 |
| **AckMessage**              | ✅ **确认消费成功**<br />消费者成功消费消息后，主动调用该接口向服务端返回消费成功响应                                                                                                                           | 🚫 无                                                                                                                                                |
| **ChangeInvisibleDuration** | 🔄 **调整处理时长**<br />消费重试场景下，修改消息处理时长，控制消息的重试间隔                                                                                                                                   | ⏱️ **消费不可见时间**<br />修改 ReceiveMessage 接口预设的消费不可见时间参数值，用于需要延长消息处理时长的场景                                        |

**可靠性重试**

SimpleConsumer 消费者类型中，客户端 SDK 和服务端通过 ReceiveMessage 和 AckMessage 接口通信。客户端 SDK 如果处理消息成功则调用 AckMessage 接口；如果处理失败只需要不回复 ACK 响应，即可在定义的消费不可见时间到达后触发消费重试流程。更多信息，请参见 SimpleConsumer 消费重试策略。

**顺序性保障**

基于 Apache RocketMQ 顺序消息的定义，SimpleConsumer 在处理顺序消息时，会按照消息存储的先后顺序获取消息。即需要保持顺序的一组消息中，如果前面的消息未处理完成，则无法获取到后面的消息。

**适用场景**

SimpleConsumer 提供原子接口，用于消息获取和提交消费结果，相对于 PushConsumer 方式更加灵活。SimpleConsumer 适用于以下场景：

- 消息处理时长不可控：如果消息处理时长无法预估，经常有长时间耗时的消息处理情况。建议使用 SimpleConsumer 消费类型，可以在消费时自定义消息的预估处理时长，若实际业务中预估的消息处理时长不符合预期，也可以通过接口提前修改。
- 需要异步化、批量消费等高级定制场景：SimpleConsumer 在 SDK 内部没有复杂的线程封装，完全由业务逻辑自由定制，可以实现异步分发、批量消费等高级定制场景。
- 需要自定义消费速率：SimpleConsumer 是由业务逻辑主动调用接口获取消息，因此可以自由调整获取消息的频率，自定义控制消费速率。

### PullConsumer

PullConsumer（拉模式消费者） 是 RocketMQ 中一种特殊的消息消费模式。 需要业务方完全手动控制

**功能特性**

- 手动拉取指定队列的消息- 业务方完全控制拉取时机
  `PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums);`
- 手动更新消费位点 - 业务方完全控制消费位点
  `void updateConsumeOffset(MessageQueue mq, long offset);`
- 获取队列信息 - 队列级别的控制
  `Set<MessageQueue> fetchMessageQueues(String topic);`

**适用场景**

- 流处理框架集成
- 数据迁移和备份
- 特殊调度需求

**优点**

- ✅ 极致控制：完全掌控消费流程

- ✅ 灵活调度：自定义拉取策略

- ✅ 资源控制：精确控制资源使用

- ✅ 位点管理：自由管理消费进度

PullConsumer 是 RocketMQ 中最底层的消费模式，提供了完全的手动控制能力，但复杂度很高。除非有特殊的流处理框架集成、数据迁移等需求，否则推荐使用 SimpleConsumer 或 PushConsumer。

## 8. 消息过滤

消息初始化时为消息设置标签或属性，消费消息时指定标签或属性进行过滤

**消息过滤分类**

| 对比项       | Tag 标签过滤                                                 | SQL 属性过滤                                                                  |
| ------------ | ------------------------------------------------------------ | ----------------------------------------------------------------------------- |
| **过滤目标** | 🏷️ **Tag 标签**<br />仅针对消息的 Tag 标签进行过滤           | 📊 **消息属性**<br />包括用户自定义属性及系统属性<br />（Tag 是一种系统属性） |
| **过滤能力** | ✅ **精准匹配**<br />完全匹配 Tag 字符串                     | 🔍 **SQL 语法匹配**<br />支持复杂条件表达式                                   |
| **过滤语法** | `java`<br />`consumer.subscribe("Topic", "TagA \|\| TagB");` | `sql`<br />`a IS NOT NULL AND a > 10`                                         |
| **适用场景** | 🎯 **简单过滤**<br />计算逻辑简单轻量                        | 🚀 **复杂过滤**<br />计算逻辑较复杂                                           |
| **性能影响** | ⚡ **高性能**<br />过滤逻辑简单，性能开销小                  | 🐢 **较高开销**<br />SQL 解析和执行需要额外计算                               |

具体的使用方式及示例，请参见[官方文档](https://rocketmq.apache.org/zh/docs/featureBehavior/07messagefilter)的 Tag 标签过滤和 SQL 属性过滤。

**使用建议**

合理划分主题和 Tag 标签

从消息的过滤机制和主题的原理机制可以看出，业务消息的拆分可以基于主题进行筛选，也可以基于主题内消息的 Tag 标签及属性进行筛选。关于拆分方式的选择，应遵循以下原则：

- 消息类型是否一致：不同类型的消息，如顺序消息和普通消息需要使用不同的主题进行拆分，无法通过 Tag 标签进行分类。
- 业务域是否相同：不同业务域和部门的消息应该拆分不同的主题。例如物流消息和支付消息应该使用两个不同的主题；同样是一个主题内的物流消息，普通物流消息和加急物流消息则可以通过不同的 Tag 进行区分。
- 消息量级和重要性是否一致：如果消息的量级规模存在巨大差异，或者说消息的链路重要程度存在差异，则应该使用不同的主题进行隔离拆分。

## 9. 消费者负载均衡

消费者从 Apache RocketMQ 获取消息消费时，通过消费者负载均衡策略，可将主题内的消息分配给指定消费者分组中的多个消费者共同分担，提高消费并发能力和消费者的水平扩展能力。本文介绍 Apache RocketMQ 消费者的负载均衡策略。

### 功能特性

- 消息消费处理的容灾策略：您可以根据消费者负载均衡策略，明确当局部节点出现故障时，消息如何进行消费重试和容灾切换。
- 消息消费的顺序性机制：通过消费者负载均衡策略，您可以进一步了解消息消费时，如何保证同一消息组内消息的先后顺序。
- 消息分配的水平拆分策略：了解消费者负载均衡策略，您可以明确消息消费压力如何被分配到不同节点，有针对性地进行流量迁移和水平扩缩容。

### 广播消费和共享消费

在 Apache RocketMQ 领域模型中，同一条消息支持被多个消费者分组订阅，同时，对于每个消费者分组可以初始化多个消费者。您可以根据消费者分组和消费者的不同组合，实现以下两种不同的消费效果：

### 广播消费和共享消费

- 广播 消息按照消费组进行消费，同一个 topic 多个消费组，每个消费组对应一个消费客户端。
- 共享消费 消费者和消费组多对一

### 什么是消费者负载均衡

消费组内共享消费场景下，消费者分组内多个消费者共同分担消息，消息按照哪种逻辑分配给哪个消费者，就是由消费者负载均衡策略所决定的。

根据消费者类型的不同，消费者负载均衡策略分为以下两种模式：

- 消息粒度负载均衡：PushConsumer 和 SimpleConsumer 默认负载策略
- 队列粒度负载均衡：PullConsumer 默认负载策略

### 消息粒度负载均衡

**使用范围**

- 对于 PushConsumer 和 SimpleConsumer 类型的消费者，默认且仅使用消息粒度负载均衡策略。

<Alert
  description="上述说明是指5.0 SDK下，PushConsumer默认使用消息粒度负载均衡，对于3.x/4.x等Remoting协议SDK 仍然使用了队列粒度负载均衡。业务集成如无特殊需求，建议使用新版本机制。"
  type="info"
  showIcon
/>

**策略原理**

消息粒度负载均衡策略中，同一消费者分组内的多个消费者将按照消息粒度平均分摊主题中的所有消息，即同一个队列中的消息，可被平均分配给多个消费者共同消费。

消息粒度的负载均衡机制，是基于内部的单条消息确认语义实现的。消费者获取某条消息后，服务端会将该消息加锁，保证这条消息对其他消费者不可见，直到该消息消费成功或消费超时。因此，即使多个消费者同时消费同一队列的消息，服务端也可保证消息不会被多个消费者重复消费。

**顺序消息负载机制**

在顺序消息中，消息的顺序性指的是同一消息组内的多个消息之间的先后顺序。因此，顺序消息场景下，消息粒度负载均衡策略还需要保证同一消息组内的消息，按照服务端存储的先后顺序进行消费。不同消费者处理同一个消息组内的消息时，会严格按照先后顺序锁定消息状态，确保同一消息组的消息串行消费。

### 队列粒度负载均衡

**使用范围**

对于历史版本（服务端 4.x/3.x 版本）的消费者，包括 PullConsumer、DefaultPushConsumer、DefaultPullConsumer、LitePullConsumer 等，默认且仅能使用队列粒度负载均衡策略。

**策略原理**

队列粒度负载均衡策略中，同一消费者分组内的多个消费者将按照队列粒度消费消息，即每个队列仅被一个消费者消费。

队列粒度的负载均衡，基于队列数量、消费者数量等运行数据进行统一的算法分配，将每个队列绑定到特定的消费者，然后每个消费者按照取消息>提交消费位点>持久化消费位点的消费语义处理消息，取消息过程不提交消费状态，因此，为了避免消息被多个消费者重复消费，每个队列仅支持被一个消费者消费。

<Alert
  description="队列粒度负载均衡策略保证同一个队列仅被一个消费者处理，该策略的实现依赖消费者和服务端的信息协商机制，Apache RocketMQ 并不能保证协商结果完全强一致。因此，在消费者数量、队列数量发生变化时，可能会出现短暂的队列分配结果不一致，从而导致少量消息被重复处理。"
  type="info"
  showIcon
/>

**策略特点**

相对于消息粒度负载均衡策略，队列粒度负载均衡策略分配粒度较大，不够灵活。但该策略在流式处理场景下有天然优势，能够保证同一队列的消息被相同的消费者处理，对于批量处理、聚合处理更友好。

**适用场景**

队列粒度负载均衡策略适用于流式计算、数据聚合等需要明确对消息进行聚合、批处理的场景。

**使用示例**

队列粒度负载均衡策略不需要额外设置，对于历史版本（服务端 4.x/3.x 版本）的消费者类型 PullConsumer 默认启用。

具体示例代码，请访问[RocketMQ 代码库](https://github.com/apache/rocketmq/blob/develop/example/src/main/java/org/apache/rocketmq/example/simple/LitePullConsumerAssign.java)获取。

### 版本兼容性

消息粒度的负载均衡策略从 Apache RocketMQ 服务端 5.0 版本开始支持，历史版本 4.x/3.x 版本仅支持队列粒度的负载均衡策略。

当您使用的 Apache RocketMQ 服务端版本为 5.x 版本时，两种消费者负载均衡策略均支持，具体生效的负载均衡策略依客户端版本和消费者类型而定。

### 使用建议

**针对消费逻辑做消息幂等**

无论是消息粒度负载均衡策略还是队列粒度负载均衡策略，在消费者上线或下线、服务端扩缩容等场景下，都会触发短暂的重新负载均衡动作。此时可能会存在短暂的负载不一致情况，出现少量消息重复的现象。因此，需要在下游消费逻辑中做好消息幂等去重处理。

## 10. 消费进度原理

Apache RocketMQ 通过消费位点管理消费进度，本文为您介绍 Apache RocketMQ 的消费进度管理机制。

### 背景信息

Apache RocketMQ 的生产者和消费者在进行消息收发时，必然会涉及以下场景，消息先生产后订阅或先订阅后生产。这两种场景下，消费者客户端启动后从哪里开始消费？如何标记已消费的消息？这些都是由 Apache RocketMQ 的消费进度管理机制来定义的。

### 功能特性

- 消息位点（Offset）
  参考 Apache RocketMQ 主题和队列的定义，消息是按到达服务端的先后顺序存储在指定主题的多个队列中，每条消息在队列中都有一个唯一的 Long 类型坐标，这个坐标被定义为消息位点。
- 消费位点（ConsumerOffset）
  Apache RocketMQ 领域模型为发布订阅模式，每个主题的队列都可以被多个消费者分组订阅。若某条消息被某个消费者消费后直接被删除，则其他订阅了该主题的消费者将无法消费该消息。
  因此，Apache RocketMQ 通过消费位点管理消息的消费进度。每条消息被某个消费者消费完成后不会立即在队列中删除，Apache RocketMQ 会基于每个消费者分组维护一份消费记录，该记录指定消费者分组消费某一个队列时，消费过的最新一条消息的位点，即消费位点。
  当消费者客户端离线，又再次重新上线时，会严格按照服务端保存的消费进度继续处理消息。如果服务端保存的历史位点信息已过期被删除，此时消费位点向前移动至服务端存储的最小位点。
- 消费位点初始值
  消费位点初始值指的是消费者分组首次启动消费者消费消息时，服务端保存的消费位点的初始值。

### 重置消费位点

若消费者分组的初始消费位点或当前消费位点不符合您的业务预期，您可以通过重置消费位点调整您的消费进度。

**适用场景**

- 初始消费位点不符合需求：因初始消费位点为当前队列的最大消息位点，即客户端会直接从最新消息开始消费。若业务上线时需要消费部分历史消息，您可以通过重置消费位点功能消费到指定时刻前的消息。
- 消费堆积快速清理：当下游消费系统性能不足或消费速度小于生产速度时，会产生大量堆积消息。若这部分堆积消息可以丢弃，您可以通过重置消费位点快速将消费位点更新到指定位置，绕过这部分堆积的消息，减少下游处理压力。
- 业务回溯，纠正处理：由于业务消费逻辑出现异常，消息被错误处理。若您希望重新消费这些已被处理的消息，可以通过重置消费位点快速将消费位点更新到历史指定位置，实现消费回溯。

**重置功能**

Apache RocketMQ 的重置消费位点提供以下能力：

- 重置到队列中的指定位点。
- 重置到某一时刻对应的消费位点，匹配位点时，服务端会根据自动匹配到该时刻最接近的消费位点。

**使用限制**

- 重置消费位点后消费者将直接从重置后的位点开始消费，对于回溯重置类场景，重置后的历史消息大多属于存储冷数据，可能会造成系统压力上升，一般称为冷读现象。因此，需要谨慎评估重置消费位点后的影响。建议严格控制重置消费位点接口的调用权限，避免无意义、高频次的消费位点重置。
- Apache RocketMQ 重置消费位点功能只能重置对消费者可见的消息，不能重置定时中、重试等待中的消息。更多信息，请参见定时/延时消息和消费重试。

### 使用建议

**严格控制消费位点重置的权限**

重置消费位点会给系统带来额外处理压力，可能会影响新消息的读写性能。 因此该操作请在适用场景下谨慎执行，并提前做好合理性和必要性评估

## 11.消费重试

消费者出现异常，消费某条消息失败时， Apache RocketMQ 会根据消费重试策略重新投递该消息进行故障恢复。本文介绍消费重试机制的原理、版本兼容性和使用建议。

### 应用场景

Apache RocketMQ 的消费重试主要解决的是业务处理逻辑失败导致的消费完整性问题，是一种为业务兜底的策略，不应该被用做业务流程控制。建议以下消费失败场景使用重试机制：

推荐使用消息重试场景如下：

- 业务处理失败，且失败原因跟当前的消息内容相关，比如该消息对应的事务状态还未获取到，预期一段时间后可执行成功。
- 消费失败的原因不会导致连续性，即当前消息消费失败是一个小概率事件，不是常态化的失败，后面的消息大概率会消费成功。此时可以对当前消息进行重试，避免进程阻塞。

典型错误使用场景如下：

- 消费处理逻辑中使用消费失败来做条件判断的结果分流，是不合理的，因为处理逻辑已经预见了一定会大量出现该判断分支。
- 消费处理中使用消费失败来做处理速率限流，是不合理的。限流的目的是将超出流量的消息暂时堆积在队列中达到削峰的作用，而不是让消息进入重试链路。

### 应用目的

消息中间件做异步解耦时的一个典型问题是如果下游服务处理消息事件失败，如何保证整个调用链路的完整性。Apache RocketMQ 作为金融级的可靠业务消息中间件，在消息投递处理机制的设计上天然支持可靠传输策略，通过完整的确认和重试机制保证每条消息都按照业务的预期被处理。

了解 Apache RocketMQ 的消息确认机制以及消费重试策略可以帮助您分析如下问题：

- 如何保证业务完整处理消息：了解消费重试策略，可以在设计实现消费者逻辑时保证每条消息处理的完整性，避免部分消息出现异常时被忽略，导致业务状态不一致。
- 系统异常时处理中的消息状态如何恢复：帮助您了解当系统出现异常（宕机故障）等场景时，处理中的消息状态如何恢复，是否会出现状态不一致。

### 消费重试策略概述

消费重试指的是，消费者在消费某条消息失败后，Apache RocketMQ 服务端会根据重试策略重新消费该消息，超过一定次数后若还未消费成功，则该消息将不再继续重试，直接被发送到死信队列中。

**消息重试的触发条件**

- 消费失败，包括消费者返回消息失败状态标识或抛出非预期异常。
- 消息处理超时，包括在 PushConsumer 中排队超时。

**消息重试策略主要行为**

- 重试过程状态机：控制消息在重试流程中的状态和变化逻辑。
- 重试间隔：上一次消费失败或超时后，下次重新尝试消费的间隔时间。
- 最大重试次数：消息可被重试消费的最大次数。

**消息重试策略差异**

根据消费者类型不同，消息重试策略的具体内部机制和设置方法有所不同，具体差异如下：

| 消费者类型         | 重试过程状态机                                                                                          | 重试间隔                                                                          | 最大重试次数                                |
| ------------------ | ------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- | ------------------------------------------- |
| **PushConsumer**   | 🔄 **完整状态流转**<br />`已就绪` → `处理中` → `待重试` → `提交` → _`死信`_<br />(支持中间等待重试状态) | ⚙️ **元数据控制**<br />• 无序消息：阶梯间隔<br />• 顺序消息：固定间隔时间         | ⚙️ **元数据控制**<br />消费者分组创建时设定 |
| **SimpleConsumer** | 🎯 **简化状态流转**<br />`已就绪` → `处理中` → `提交` → _`死信`_<br />(无显式待重试状态)                | 🔧 **API 动态控制**<br />通过`ChangeInvisibleDuration`API<br />修改消息不可见时间 | ⚙️ **元数据控制**<br />消费者分组创建时设定 |

### 使用建议

**合理重试，避免因限流等诉求触发消费重试**

上文应用场景中提到，消息重试适用业务处理失败且当前消费为小概率事件的场景，不适合在连续性失败的场景下使用，例如消费限流场景。

- 错误示例：如果当前消费速度过高触发限流，则返回消费失败，等待下次重新消费。
- 正确示例：如果当前消费速度过高触发限流，则延迟获取消息，稍后再消费。

**合理控制重试次数，避免无限重试**

虽然 Apache RocketMQ 支持自定义消费重试次数，但是建议通过减少重试次数+延长重试间隔来降低系统压力，避免出现无限重试或大量重试的情况。

## 12.消息存储和清理机制

本文为您介绍 Apache RocketMQ 中消息的存储机制，包括消息的存储粒度、判断依据及后续处理策略等。

### 背景信息

参考 Apache RocketMQ 中队列的定义，消息按照达到服务器的先后顺序被存储到队列中，理论上每个队列都支持无限存储。

但是在实际部署场景中，服务端节点的物理存储空间有限，消息无法做到永久存储。因此，在实际使用中需要考虑以下问题，消息在服务端中的存储以什么维度为判定条件？消息存储以什么粒度进行管理？消息存储超过限制后如何处理？这些问题都是由消息存储和过期清理机制来定义的。

了解消息存储和过期清理机制，可以从以下方面帮助您更好的进行运维管理：

- 提供消息存储时间 SLA，为业务提供安全冗余空间：消息存储时间的承诺本质上代表业务侧可以自由获取消息的时间范围。对于消费时间长、消息堆积、故障恢复等场景非常关键。
- 评估和控制存储成本：Apache RocketMQ 消息一般存储于磁盘介质上，您可以通过存储机制评估消息存储空间，提前预留存储资源。

### 消息存储机制

**原理机制**

Apache RocketMQ 使用存储时长作为消息存储的依据，即每个节点对外承诺消息的存储时长。在存储时长范围内的消息都会被保留，无论消息是否被消费；超过时长限制的消息则会被清理掉。

消息存储机制主要定义以下关键问题：

- 消息存储管理粒度：Apache RocketMQ 按存储节点管理消息的存储时长，并不是按照主题或队列粒度来管理。
- 消息存储判断依据：消息存储按照存储时间作为判断依据，相对于消息数量、消息大小等条件，使用存储时间作为判断依据，更利于业务方对消息数据的价值进行评估。
- 消息存储和是否消费状态无关：Apache RocketMQ 的消息存储是按照消息的生产时间计算，和消息是否被消费无关。按照统一的计算策略可以有效地简化存储机制。

**消息存储管理粒度说明**

Apache RocketMQ 按照服务端节点粒度管理存储时长而非队列或主题，原因如下：

- 消息存储优势权衡：Apache RocketMQ 基于统一的物理日志队列和轻量化逻辑队列的二级组织方式，管理物理数据。这种机制可以带来顺序读写、高吞吐、高性能等优势，但缺点是不支持按主题和队列单独管理。
- 安全生产和容量保障风险要求：即使 Apache RocketMQ 按照主题或者队列独立生成存储文件，但存储层本质还是共享存储介质。单独根据主题或队列控制存储时长，这种方式看似更灵活，但实际上整个集群仍然存在容量风险，可能会导致存储时长 SLA 被打破。从安全生产角度考虑，最合理的方式是将不同存储时长的消息通过不同集群进行分离治理。

**消息存储和消费状态关系说明**

Apache RocketMQ 统一管理消息的存储时长，无论消息是否被消费。

当消费者不在线或消息消费异常时，会造成队列中大量消息堆积，且该现象暂时无法有效控制。若此时按照消费状态考虑将未消费的消息全部保留，则很容易导致存储空间不足，进而影响到新消息的读写速度。

根据统一地存储时长管理消息，可以帮助消费者业务清晰地判断每条消息的生命周期。只要消息在有效期内可以随时被消费，或通过重置消费位点功能使消息可被消费多次。

### 消息过期清理机制

在 Apache RocketMQ 中，消息保存时长并不能完整控制消息的实际保存时间，因为消息存储仍然使用本地磁盘，本地磁盘空间不足时，为保证服务稳定性消息仍然会被强制清理，导致消息的实际保存时长小于设置的保存时长。

### 使用建议

**消息存储时长建议适当增加**

Apache RocketMQ 按存储时长统一控制消息是否保留。建议在存储成本可控的前提下，尽可能延长消息存储时长。延长消息存储时长，可以为紧急故障恢复、应急问题排查和消息回溯带来更多的可操作空间。
